package com.k2data.service;

import java.util.HashMap;
import java.util.Map;

import k2paltform.common.util.PathWatcher;
import org.apache.avro.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * load all schema from Zookeeper and provide a function to get schema according Schema ID.
 * @author Ming
 *
 */
public class ZkSchema extends PathWatcher<FieldGroup> {
	public static final Logger LOG = LoggerFactory.getLogger(ZkSchema.class);

	public ZkSchema() {
		super();
	}

	public ZkSchema(String basePath) {
		super(basePath);
	}

	@Override
	public FieldGroup getDataHolder(byte[] data) {
		try{
			return new FieldGroup(data);
		} catch (Exception e) {
			LOG.warn("Exception when parsing schema.", e);
			return null;
		}
	}

	public Schema getSchema(String schemaID) {
		if(this.getChildData(schemaID) == null)
			return null;
		return this.getChildData(schemaID).getSchema();
	}

	public String[] getIdFields(String schemaID) {
		if(this.getChildData(schemaID) == null)
			return null;
		return this.getChildData(schemaID).getIdFields();
	}

	public FieldGroup getFieldGroup(String schemaID) {
		return this.getChildData(schemaID);
	}

	public Map<String, FieldGroup> getFieldGroups() {
		return this.getChildrenData();
	}

	public Map<String, Schema> getSchemas() {
		Map<String, Schema> schemas = new HashMap<>();
		for(String schemaId : this.getChildrenData().keySet()) {
			schemas.put(schemaId, this.getSchema(schemaId));
		}
		return schemas;
	}

	public Map<String, String[]> getIdFields() {
		Map<String, String[]> idFields = new HashMap<>();
		for(String schemaId : this.getChildrenData().keySet()) {
			idFields.put(schemaId, this.getIdFields(schemaId));
		}
		return idFields;
	}

	public void clearSchemas() {
		this.clear();
	}
}
